package com.o2o.cleaning.month.platform.ebusiness_plat.brand_modular.utils

import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.spark.sql.functions.lit
import org.apache.spark.sql.{DataFrame, SparkSession}

class Fangshai {
  def fangshai(brandEndResult: DataFrame,spark:SparkSession) = {
//for (i <- Array("jumei", "yunji", "miya", "guomei", "mryx", "intime", "palmlife","aomygad","womaiwang","rsmsh","rongyigou","beidian","yangmatou","quchenshi")){
//  if(i.equals(platform)){
//
//  }
//}
    val result = brandEndResult.toJSON.rdd.map(line => {
      val jsons = JSON.parseObject(line)
      val thirdCategoryId = jsons.getOrDefault("thirdCategoryId", "-1")
      if (thirdCategoryId.equals("100190106")) {
        val base_info = jsons.getOrDefault("Base_Info", "-1").toString
        val title = jsons.getOrDefault("title", "-1").toString
        // 有base_info的从base_info中清洗
        if (!base_info.toString.equals("-1")) {
          val base = jsons.getJSONObject("Base_Info")
          val fangshai = base.getOrDefault("SPF值", "-1").toString
          // base_info中有防晒指数的
          val sp_fangshai = fangshai.toUpperCase()
            .replaceAll("\\+", "")
            .replaceAll(" ", "")
            .replaceAll("PA", "")
            .replaceAll("/", "")
            .replaceAll("SPF", "")
            .replaceAll("spf", "")
            .replaceAll("\\(", "")
            .replaceAll("\\（", "")
            .replaceAll("【", "")
            .replaceAll("\\.", "")
            .replaceAll(":", "")
            .replaceAll("""[\u4e00-\u9fa5 ]""", "")

          if (fangshai.toUpperCase().contains("SPF") && !sp_fangshai.contains("-")) {
            jsons.put("SPF", sp_fangshai)
          } else {
            title_sp(title, jsons)
          }
        } else {
          title_sp(title, jsons)
        }
      }
      jsons.toString
    })

    val result_end = spark.read.json(result)
      .selectExpr("*", "case when SPF = null or SPF = '' or SPF rlike '其他' or SPF rlike '无' then '-1'  else SPF  end SPF_1")
      .drop("SPF")
      .withColumnRenamed("SPF_1", "SPF")

    val brandNameJosn = result_end.toJSON.rdd.map(line => {
      val lines = JSON.parseObject(line)
      val SPF = lines.getOrDefault("SPF", "-1").toString
      if (SPF.replaceAll("[0-9]+", "").length == 0) {
        lines.put("SPF", SPF)
      } else {
        lines.put("SPF", "-1")
      }
      lines.remove("_id")
      lines
    })
    brandNameJosn
  }

  def title_sp(title: String, jsons: JSONObject) = {

    if (title.toUpperCase.contains("SPF")) {
      val etl_title = title.replaceAll(" ", "").toUpperCase()
      val bb = etl_title.indexOf("SPF")
      if (bb != etl_title.length - 3 && bb < etl_title.length && bb + 5 < etl_title.length) {
        val cc = etl_title.substring(bb, bb + 5)
        if (cc.toUpperCase().contains("SPF")) {
          val cc_fangshai = cc.toUpperCase()
            .replaceAll("\\+", "")
            .replaceAll(" ", "")
            .replaceAll("PA", "")
            .replaceAll("/", "")
            .replaceAll("SPF", "")
            .replaceAll("spf", "")
            .replaceAll("\\(", "")
            .replaceAll("\\（", "")
            .replaceAll("【", "")
            .replaceAll("\\.", "")
            .trim
          if (cc_fangshai.isEmpty) {
            jsons.put("SPF", "-1")
          } else {
            jsons.put("SPF", cc_fangshai)
          }


        } else {
          jsons.put("SPF", "-1")
        }
      } else {
        jsons.put("SPF", "-1")
      }
    } else {
      jsons.put("SPF", "-1")
    }
  }
}
